package cn.edu.tsinghua.mutimedia.format;

import java.io.IOException;




import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;


/**
 * An input format that reads the first 10 characters of each line as the key
 * and the rest of the line as the value. Both key and value are represented
 * as Text.
 */
public class ImageInputFormat extends  FileInputFormat<Text, Text> {
	    
	@Override
    public List<InputSplit> getSplits(JobContext context) throws IOException {
        List<InputSplit> splits = new ArrayList<InputSplit>();
		List<FileStatus> files = listStatus(context);
        for (FileStatus file: files) {
            Path path = file.getPath();
            System.out.println("ImageInputFormat:******************************path:"+path.toString()+"*******************************");
            long length = file.getLen();
            if (length != 0) {
                FileSystem fs = path.getFileSystem(context.getConfiguration());
                BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
                splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
            } else {
                //Create empty hosts array for zero length files
                splits.add(makeSplit(path, 0, length, new String[0]));
            }
        }
        return splits;
    }

    @Override
    public RecordReader<Text, Text> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new ImageFileRecordReader((FileSplit)split,context,true);
    }


    /**
     * A factory that makes the split for this class. It can be overridden
     * by sub-classes to make sub-types
     */
    protected FileSplit makeSplit(Path file, long start, long length,
                                  String[] hosts) {
        return new FileSplit(file, start, length, hosts);
    }
}
